package cn.tedu.charging.order.emqx;

import cn.tedu.charging.common.pojo.entity.ProgressEntity;
import cn.tedu.charging.common.pojo.message.ChargingCheckResultMessage;
import cn.tedu.charging.common.pojo.message.ChargingProgressMsg;
import cn.tedu.charging.common.pojo.message.ProgressData;
import cn.tedu.charging.common.pojo.param.ProgressCostParam;
import cn.tedu.charging.common.pojo.po.ChargingBillFailPO;
import cn.tedu.charging.common.pojo.po.ChargingBillSuccessPO;
import cn.tedu.charging.common.pojo.vo.ProgressCostVO;
import cn.tedu.charging.common.protocol.JsonResult;
import cn.tedu.charging.common.protocol.WebSocketMessage;
import cn.tedu.charging.order.dao.mapper.ChargingBillSuccessMapper;
import cn.tedu.charging.order.dao.repository.ProgressESRepository;
import cn.tedu.charging.order.feign.CostClient;
import cn.tedu.charging.order.websocket.WebSocketServer;
import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import groovyjarjarpicocli.CommandLine;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.BeanUtils;
import org.springframework.data.elasticsearch.core.query.UpdateQuery;

import java.nio.charset.StandardCharsets;
import java.util.Date;

@Slf4j
public class MqttConsumer implements MqttCallbackExtended {
    private MqttClient mqttClient;
    private MqttContext mqttContext;


    public MqttConsumer(MqttClient mqttClient, MqttContext mqttContext) {
        this.mqttClient = mqttClient;
        this.mqttContext = mqttContext;
       /* this.orderMapper = orderMapper;*/
    }

    /**
     * 连接成功 客户端mqttClient成功的建立了和emqx连接
     * @param reconnect mqttClient 回调消费传递的参数 告知 方法 本次连接建立是否支持重连
     * @param serverURI 服务端emqx地址
     */
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        //客户端对象mqttClient成功连接的时候 调用的 通过客户端mqttClient实现监听订阅
        try {
            mqttClient.subscribe("/charging/check/result");
            mqttClient.subscribe("/charging/progress");
        } catch (MqttException e) {
            log.error("订阅主题失败",e);
        }
    }

    /**
     * 连接断开 mqttClient断开和服务连接 回调的方法
     * 消费者使用的一些资源 消耗程序内容 在连接断开时候 释放避免因为这个消费者无法使用消费逻辑导致程序阻塞
     * @param cause the reason behind the loss of connection.
     */
    @Override
    public void connectionLost(Throwable cause) {
        log.error("连接断开",cause);
    }

    /**
     * 消费者通过mqttClient监听订阅的主题 产生了消息,消息到达了当前消费对象
     * @param topic 消息主题来源 主题名称
     * @param message 消息对象
     * @throws Exception
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.debug("主题:{},消息:{}",topic,message);
        //如果消息来源主题不一样 消费逻辑不同
        if (topic!=null&&topic.equals("/charging/check/result")){
            //当前消费者 绑定的连接对象监听多个主题,如果消息来源是 反馈设备自检结果的主题,处理自检结果
            handleCheckResult(topic,message);
        }else if(topic!=null&&topic.equals("/charging/progress")){
            //当前消费者 绑定的连接对象监听多个主题,如果消息来源是 充电进度主题,处理进度同步
            handleChargingProgress(topic,message);
        }
    }

    private void handleChargingProgress(String topic, MqttMessage message) {
        //充电进度消费逻辑
        log.debug("当前消费者,主题:{},获取到消息:{},开始执行充电进度消费逻辑",topic,message);
        //1.解析消息对象 转化成message数据封装类
        String jsonMsg=new String(message.getPayload(), StandardCharsets.UTF_8);
        ChargingProgressMsg progress=JSON.parseObject(jsonMsg, ChargingProgressMsg.class);
        //2.UNDO 根据消息中传递的温度数据,判断是否安全,如果不安全 停止充电 通知手机操作
        checkTemperature(progress);
        //3.远程调用计价中心,计算本次充电价格数据,返回ProgressCostVO计价vo对象完成后续流程
        ProgressCostVO vo=calculate(progress);
        //4.封装一个ES的实体,记录充电进度信息,私用es客户端Repository 写入es存储
        saveProgress(progress,vo.getTotalCost());
        //5.判断是否结束 结束 没结束 反馈给前端webSocket不一样 结束 state=4 没结束 组织webSocket 通知前端
        Boolean isFull = progress.getIsFull();
        if (isFull){
            //状态修改
            ChargingBillSuccessMapper chargingBillSuccessMapper = mqttContext.getChargingBillSuccessMapper();
            UpdateWrapper wrapper=new UpdateWrapper();
            wrapper.eq("bill_id",progress.getOrderNo());
            wrapper.setSql("bill_status=2");
            chargingBillSuccessMapper.update(null,wrapper);
            //通知用户 充电最终金额 totalCost 充电结束 webSocketMessage.state=1 前端不展示
            WebSocketServer webSocketServer = mqttContext.getWebSocketServer();
            WebSocketMessage<String> webSocketMessage = new WebSocketMessage<>();
            webSocketMessage.setState(1);
            webSocketMessage.setMessage("over");
            webSocketMessage.setData("A 当前订单充电结束,请支付离开");
            webSocketServer.pushMessage(progress.getUserId(),JSON.toJSONString(webSocketMessage));
            //通知订单数据 success status= 2完成 1充电中 2完成 3异常
            //update charging_bill_success set status=2 where bill_id=#{orderNo}
//            UpdateQuery
//            mqttContext.getChargingBillSuccessMapper().update
        }else{
            //通知用户 充电的进度显示 totalCost state=3   1 错误 A开头 异常 B开头订单创建失败  2 余额不足  3展示充电进度
            //通知用户 充电最终金额 totalCost 充电结束 webSocketMessage.state=1 前端不展示
            WebSocketServer webSocketServer = mqttContext.getWebSocketServer();
            WebSocketMessage<ProgressData> webSocketMessage = new WebSocketMessage<>();
            ProgressData data=new ProgressData();
            //设备传递的充电度数
            data.setChargingCapacity(progress.getChargingCapacity());
            //可以使用redisTemplate 做incrby RedisTemplate 不行必须支持value string序列化StringRedisTemplate
            data.setTotalCapacity(100.00);
            //设备充电 目前单价 和总金额
            data.setOneElectricityCost(vo.getPowerFee());
            data.setTotalCost(vo.getTotalCost());
            data.setHours(0);
            data.setMinutes(10);
            data.setSeconds(10);
            webSocketMessage.setState(3);
            webSocketMessage.setMessage("charging");
            webSocketMessage.setData(data);
            webSocketServer.pushMessage(progress.getUserId(),JSON.toJSONString(webSocketMessage));
        }
    }
    private void saveProgress(ChargingProgressMsg progress, Double totalCost) {
        //1.从上下文里 拿到 repository
        ProgressESRepository progressESRepository = mqttContext.getProgressESRepository();
        //2.调用的是新增数据到es post 新增文档 组织一个文档对象
        ProgressEntity entity=new ProgressEntity();
        //绝大部分 来自于 设备同步消息
        BeanUtils.copyProperties(progress,entity);
        //总金额 来自于 计价中心
        entity.setTotalCost(totalCost);
        //时序id 使用雪花id算法生成 时间戳(简陋了) 有可能同一时间点毫秒数 多个订单都在同步充电进度
        //每张订单,同一张订单,会不会有时间戳冲突 不会
        String id=progress.getOrderNo()+System.currentTimeMillis();
        entity.setId(id);
        progressESRepository.save(entity);
    }

    private ProgressCostVO calculate(ChargingProgressMsg progress) {
        log.debug("订单解析消息,调用计价中心,计算价格,本次订单:{}",progress.getOrderNo());
        //1.从mqttContext拿到 costClient
        CostClient costClient = mqttContext.getCostClient();
        //2.使用消息progress 直接拷贝 param 入参
        ProgressCostParam param=new ProgressCostParam();
        BeanUtils.copyProperties(progress,param);
        //3.调用请求 解析结果 返回
        JsonResult<ProgressCostVO> result = costClient.calculateCost(param);
        log.debug("远程调用计价中心,计算价格,返回结果:{}",result);
        return result.getData();
    }

    private boolean checkTemperature(ChargingProgressMsg progress) {
        return true;
    }

    private void handleCheckResult(String topic, MqttMessage message) {
        log.debug("当前消费者,主题:{},获取到消息:{},开始处理设备自检反馈",topic,message);
        log.debug("先抢锁");
        //1.解析message body string --> json --> 对象 判断true false
        String msgJson=new String(message.getPayload(), StandardCharsets.UTF_8);
        ChargingCheckResultMessage msg = JSON.parseObject(msgJson, ChargingCheckResultMessage.class);
        Boolean result = msg.getResult();
        //2.根据判断结果 调用mqttContext封装不同fail success po写入数据库
        if (result){
            //SUCCESS
            ChargingBillSuccessPO po = new ChargingBillSuccessPO();
            po.setBillId(msg.getOrderNo());
            po.setUserId(msg.getUserId());
            po.setBillStatus(1);//status状态 成功创建后 需要初始化
            po.setCreateTime(new Date());
            //如果需要别的,当前消息数据获取来源只能是设备充电桩 丰富一下消息内容
            mqttContext.getChargingBillSuccessMapper().insert(po);
            //UNDO 发消息 告知充电桩可以真正开始充电 然后同步给我充点进度
        }else{
            //FAIL
            ChargingBillFailPO po = new ChargingBillFailPO();
            po.setBillId(msg.getOrderNo());
            po.setUserId(msg.getUserId());
            po.setCreateTime(new Date());
            po.setFailDesc("设备自检反馈失败");
            mqttContext.getChargingBillFailMapper().insert(po);
            //给用户生成一个优惠券 webSocket 通知用户 换枪领券
            WebSocketMessage<String> webSocketMessage = new WebSocketMessage<>();
            webSocketMessage.setState(1);//1 失败 异常  2 余额不足  3 正常进度同步
            webSocketMessage.setMessage("fail");
            webSocketMessage.setData("B 您的订单创建失败,设备有故障,对不起,送您一张优惠券");
            mqttContext.getWebSocketServer()
                    .pushMessage(msg.getUserId(), JSON.toJSONString(webSocketMessage));
        }
    }

    /**
     * 消息投递 消息消费彻底完成
     * @param token message身份标识
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.debug("消息投递完成");
    }
}
